package com.monitor.zkm.core;

import bean.ZkConfig;
import com.monitor.zkm.core.listener.PathAgentListener;
import com.monitor.zkm.dao.HostStatusRepository;
import com.monitor.zkm.util.FieldUtils;
import core.scan.ZKContains;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.*;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.framework.state.ConnectionStateListener;
import org.apache.curator.retry.RetryNTimes;
import org.apache.zookeeper.CreateMode;

import javax.annotation.PreDestroy;
import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.List;

@Slf4j
public class ZkClient {

    private CuratorFramework curatorFramework;
    private ZkConfig zkConfig;
    private final List<Object> excludeField = new ArrayList<>();
    private final HostStatusRepository hostStatusRepository;

    public ZkClient(ZkConfig conf, HostStatusRepository hostStatusRepository) throws Exception{
        try {
            if(checkZkConfig(conf)){
                zkConfig = conf;
            }
            this.hostStatusRepository = hostStatusRepository;
        } catch (Exception e) {
            log.error(e.getMessage());
            throw new Exception(e);
        }
    }

    public void start() throws Exception {
        RetryNTimes retryNTimes = new RetryNTimes(zkConfig.getRetryNum(),zkConfig.getRetryOnceTime());

        curatorFramework = CuratorFrameworkFactory.newClient(
                zkConfig.getIpAddr(),
                zkConfig.getSessionTimeOut(),
                zkConfig.getConnectTimeOut(),
                retryNTimes
        );

        curatorFramework.start();

        initCreateNode();
        initListen();
    }

    public void initListen() throws Exception {
        addConnectionListener();
        registNodeListen();
    }

    public void initCreateNode() throws Exception {
        String localAddress = new String(CuratorFrameworkFactory.getLocalAddress());
        String newNodePath = ZKContains.CLIENT_NODE + "/" + localAddress;
        if(curatorFramework.checkExists().forPath(newNodePath) == null){
            curatorFramework.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL)
            .forPath(newNodePath,localAddress.getBytes());
        }
    }


    protected boolean checkZkConfig(ZkConfig config) throws Exception {
        if(config == null){
            throw new Exception("config is null");
        }

        Field[] fields = config.getClass().getDeclaredFields();
        for(Field field : fields){
            field.setAccessible(true);
            if(excludeField.contains(field.getName())){
                continue;
            }
            FieldUtils.checkIsNotNullFieldByType(field,config);
        }
        return true;
    }

    /**
     * only master can regist listen
     */
    private void registNodeListen() throws Exception {
        PathChildrenCache childrenCache = new PathChildrenCache(curatorFramework,ZKContains.AGENT_NODE,true);
        childrenCache.getListenable().addListener(new PathAgentListener(hostStatusRepository));
        childrenCache.start();
    }


    /**
     * add listener
     */
    private void addConnectionListener(){
        ConnectionStateListener connectionStateListener = new ConnectionStateListener() {
            public void stateChanged(CuratorFramework curatorFramework, ConnectionState connectionState) {
                log.info("state changed , current state : " + connectionState.name());
                /**
                 * probably session expired
                 */
                if(connectionState == ConnectionState.LOST){
                    log.info("current zookeepr connection state : connection lost ");
                    while(true){
                        try {
                            if(curatorFramework.getZookeeperClient().blockUntilConnectedOrTimedOut()){
                                initCreateNode();
                                log.info("重连zk成功");
                                break;
                            }else{
                                curatorFramework.start();
                            }
                            Thread.sleep(10000);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                            break;
                        } catch (Exception e){
                            e.printStackTrace();
                        }
                    }
                }else if(connectionState == ConnectionState.SUSPENDED){
                    log.error("session 超时");
                    curatorFramework.start();
                    try {
                        if(curatorFramework.getZookeeperClient().blockUntilConnectedOrTimedOut()){
                            initCreateNode();
                            log.info("重连zk成功");
                        }
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }else if(connectionState == ConnectionState.CONNECTED){
                    log.info("connect zookeeper success");
                }else if(connectionState == ConnectionState.RECONNECTED){
                    log.info("重新连接zk");
                }
            }
        };

        curatorFramework.getConnectionStateListenable().addListener(connectionStateListener);
    }

    @PreDestroy
    public void close(){
        if(curatorFramework != null){
            curatorFramework.close();
        }
    }

    /**
     * 刷新配置
     */
    public void reflush(ZkConfig zkConfig) throws Exception {
        close();
        this.zkConfig = zkConfig;
        start();
    }
}
